package com.shujia.hbase.api;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.text.SimpleDateFormat;

public class Demo9 {
    private HConnection connection;
    private HBaseAdmin hBaseAdmin;

    @Before
    public void init() {

        //创建hadoop配置文件对象
        Configuration configuration = new Configuration();

        //指定zk地址
        configuration.set("hbase.zookeeper.quorum", "node1:2181,node2:2181,node3:2181");


        try {
            //建立连接   和zk的连接
            connection = HConnectionManager.createConnection(configuration);

            hBaseAdmin = new HBaseAdmin(configuration);

            System.out.println("连接建立成功");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    @Test
    public void put() {

        /**
         * create 'dianxin','info'
         *
         */

        try {
            HTableInterface dianxin = connection.getTable("dianxin");
            SimpleDateFormat format = new SimpleDateFormat("yyyyMMddHHmmss");

            BufferedReader bufferedReader = new BufferedReader(new FileReader("data/dianxin_data"));

            String line;
            while ((line = bufferedReader.readLine()) != null) {
                String[] split = line.split(",");
                if (!"\\N".equals(split[1])) {

                    String date = split[5];

                    //获取时间戳
                    long time = format.parse(date).getTime();


                    //手机号
                    String id = split[0];


                    String rowkey = id + "_" + (Long.MAX_VALUE - time);


                    Put put = new Put(rowkey.getBytes());
                    put.add("info".getBytes(), "city".getBytes(), split[2].getBytes());
                    put.add("info".getBytes(), "grid".getBytes(), split[1].getBytes());
                    put.add("info".getBytes(), "date".getBytes(), split[7].getBytes());


                    dianxin.put(put);

                }


            }


        } catch (Exception e) {
            e.printStackTrace();
        }


    }


    @Test
    public void query1() {

        //通过手机号查询用户最近10条位置记录
        try {

            long start = System.currentTimeMillis();


            String id = "BB3B0B664A3CFC5D0580056347E1D3E156C0A919";

            HTableInterface dianxin = connection.getTable("dianxin");

            //前缀比较器
            BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator(id.getBytes());

            RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, binaryPrefixComparator);

            PageFilter pageFilter = new PageFilter(3);


            FilterList filterList = new FilterList();

            filterList.addFilter(rowFilter);

            filterList.addFilter(pageFilter);

            Scan scan = new Scan();

            scan.setFilter(filterList);


            ResultScanner scanner = dianxin.getScanner(scan);

            Result result;

            while ((result = scanner.next()) != null) {

                String city = Bytes.toString(result.getValue("info".getBytes(), "city".getBytes()));
                String grid = Bytes.toString(result.getValue("info".getBytes(), "grid".getBytes()));
                String date = Bytes.toString(result.getValue("info".getBytes(), "date".getBytes()));

                System.out.println(city + "\t" + grid + "\t" + date);

            }


            long end = System.currentTimeMillis();
            System.out.println(end - start);


        } catch (IOException e) {
            e.printStackTrace();
        }

    }


    @Test
    public void query2() {

        //获取用户某一天在一个城市中的所有位置
        try {

            long start = System.currentTimeMillis();


            String id = "7C7F09D9FB218A2F812AFD0DF8C71B1303F17AF0";

            String date = "20180503";

            String city = "83401";


            HTableInterface dianxin = connection.getTable("dianxin");

            //前缀比较器
            BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator(id.getBytes());

            RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, binaryPrefixComparator);

            BinaryComparator cityCom = new BinaryComparator(city.getBytes());
            SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter("info".getBytes(), "city".getBytes(),
                    CompareFilter.CompareOp.EQUAL, cityCom);


            BinaryComparator dateCom = new BinaryComparator(date.getBytes());
            SingleColumnValueFilter datesingleColumnValueFilter = new SingleColumnValueFilter("info".getBytes(), "date".getBytes(),
                    CompareFilter.CompareOp.EQUAL, dateCom);


            FilterList filterList = new FilterList();

            filterList.addFilter(singleColumnValueFilter);
            filterList.addFilter(datesingleColumnValueFilter);
            filterList.addFilter(rowFilter);


            Scan scan = new Scan();
            scan.setFilter(filterList);

            ResultScanner scanner = dianxin.getScanner(scan);

            Result result;

            while ((result = scanner.next()) != null) {

                String city1 = Bytes.toString(result.getValue("info".getBytes(), "city".getBytes()));
                String grid = Bytes.toString(result.getValue("info".getBytes(), "grid".getBytes()));
                String date1 = Bytes.toString(result.getValue("info".getBytes(), "date".getBytes()));

                System.out.println(city1 + "\t" + grid + "\t" + date1);

            }


            long end = System.currentTimeMillis();
            System.out.println(end - start);


        } catch (IOException e) {
            e.printStackTrace();
        }

    }


}
